package com.sailing.lianxi.service;

import java.util.Collection;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;

import com.yammer.metrics.stats.EWMA;

import oracle.net.aso.n;
import oracle.net.aso.p;


/**
 * 自定义负载均衡监听器
 * @author Administrator
 *
 */
public class MyConsumerRebalancerListener implements ConsumerRebalanceListener  {

	
	Consumer<String, String> consumer;
	MessageDao messageDao;
	
	public MyConsumerRebalancerListener(Consumer<String, String> consumer,MessageDao messageDao) {
		this.consumer = consumer;
		this.messageDao = messageDao;
	}
	
	  //任务被取消
	public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
		// TODO Auto-generated method stub
		for (TopicPartition topicPartition : partitions) {
			long offset = consumer.position(topicPartition);
			MessageOffsetPO po = new MessageOffsetPO();
			po.setOffset(offset);
			po.setPartition(topicPartition.partition()+"");
			po.setTopic(topicPartition.topic());
			try {
				messageDao.insert(po);
			} catch (Exception e) {
				// TODO: handle exception
			}
			System.out.printf("onPartitionsRevoked topic:%s,messageOffsetPO:%s",topicPartition.topic(),po);
		}
	}

	//收到新任务
	public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
		// TODO Auto-generated method stub
		for (TopicPartition topicPartition : partitions) {
			MessageOffsetPO po = messageDao.get(topicPartition.topic(), topicPartition.partition());
			if(null == po) {//接收到新的topic，即数据库中记录不存在
				consumer.seek(topicPartition, 0);
			}else {
				consumer.seek(topicPartition, po.getOffset()+1);//下一个offset，所以需要加1
			}
			System.out.printf("onPartitionsAssigned topic:%s,messageOffsetPO:%s,offset:%d", topicPartition.topic(),po.getOffset());
		}
	}

	

}
